package com.sec.etech.redis.stream.consumer.group;


import com.sec.etech.redis.stream.RedisStreamUtil;
import com.sec.etech.redis.stream.consumer.CustomErrorHandler;
import com.sec.etech.redis.stream.consumer.TestAsyncConsumeStreamListener;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.data.redis.stream.Subscription;

import javax.annotation.Resource;
import java.time.Duration;

@Slf4j
//@Configuration
public class RedisStreamConfiguration {

    @Resource
    private RedisStreamUtil redisStreamUtil;
    //@Resource
    //private TestAsyncConsumeStreamListener listener;

    @Bean
    public StreamMessageListenerContainer.StreamMessageListenerContainerOptions streamMessageListenerContainerOptions() {
        return StreamMessageListenerContainer
                .StreamMessageListenerContainerOptions
                .builder()
                // 一次最多获取多少条消息
                .batchSize(3)
                // 运行 Stream 的 poll task
                //.executor(executor)
                // 可以理解为 Stream Key 的序列化方式
                .keySerializer(RedisSerializer.string())
                // 可以理解为 Stream 后方的字段的 key 的序列化方式
                .hashKeySerializer(RedisSerializer.string())
                // 可以理解为 Stream 后方的字段的 value 的序列化方式
                .hashValueSerializer(RedisSerializer.string())
                // Stream 中没有消息时，阻塞多长时间，需要比 `spring.redis.timeout` 的时间小 秒
                .pollTimeout(Duration.ofSeconds(3))
                // ObjectRecord 时，将 对象的 filed 和 value 转换成一个 Map 比如：将Book对象转换成map
                //.objectMapper(new ObjectHashMapper())
                // 获取消息的过程或获取到消息给具体的消息者处理的过程中，发生了异常的处理
                .errorHandler(new CustomErrorHandler())
                // 将发送到Stream中的Record转换成ObjectRecord，转换成具体的类型是这个地方指定的类型
                //.targetType(XXX.class)
                .build();
    }

    @Bean
    public StreamMessageListenerContainer streamMessageListenerContainer(RedisConnectionFactory factory,
                                                                         StreamMessageListenerContainer.StreamMessageListenerContainerOptions streamMessageListenerContainerOptions) {
        StreamMessageListenerContainer listenerContainer = StreamMessageListenerContainer.create(factory,
                streamMessageListenerContainerOptions);
        listenerContainer.start();
        return listenerContainer;
    }

    /**
     * 订阅者1，消费组group1，收到消息后不自动确认，与订阅者2为竞争关系，消息仅被其中一个消费
     *
     * @param streamMessageListenerContainer
     * @return subscription
     */
    /*@Bean
    public Subscription subscription(StreamMessageListenerContainer streamMessageListenerContainer) {
        String redisStreamKey = "test_stream";
        redisStreamUtil.initGroup(redisStreamKey,  "group-a");//从头消费
        Subscription subscription = streamMessageListenerContainer.receive(Consumer.from("group-a", "consumer-a"),
                StreamOffset.create(redisStreamKey, ReadOffset.lastConsumed()),
                new TestAsyncConsumeStreamListener("消费组A消费", "group-a", "consumer-a"));
        return subscription;
    }*/

    /**
     * 订阅者2，消费组group1，收到消息后自动确认，与订阅者1为竞争关系，消息仅被其中一个消费
     *
     * @param streamMessageListenerContainer
     * @return subscription
     */
    /*@Bean
    public Subscription subscription2(StreamMessageListenerContainer streamMessageListenerContainer) {
        String redisStreamKey = "test_stream";
        redisStreamUtil.initGroup(redisStreamKey, "group-a");//从头消费
        Subscription subscription = streamMessageListenerContainer.receiveAutoAck(Consumer.from("group-a", "consumer-b"),
                StreamOffset.create(redisStreamKey, ReadOffset.lastConsumed()),
                new TestAsyncConsumeStreamListener("消费组A自动确认消费", "group-a", "consumer-b"));
        return subscription;
    }*/

    /**
     * 订阅者3，消费组group2，收到消息后不自动确认，需要用户选择合适的时机确认，与订阅者1和2非竞争关系，即使消息被订阅者1或2消费，亦可消费
     * <p>
     * 当某个消息被ACK，PEL列表就会减少
     * 如果忘记确认（ACK），则PEL列表会不断增长占用内存
     * 如果服务器发生意外，重启连接后将再次收到PEL中的消息ID列表
     *
     * @param factory
     * @return subscription
     */
    /*@Bean
    public Subscription subscription3(RedisConnectionFactory factory) {
        StreamMessageListenerContainer.StreamMessageListenerContainerOptions options = StreamMessageListenerContainer
                .StreamMessageListenerContainerOptions
                .builder()
                // 一次最多获取多少条消息
                .batchSize(3)
                .pollTimeout(Duration.ofSeconds(3))
                .build();

        StreamMessageListenerContainer listenerContainer = StreamMessageListenerContainer.create(factory, options);
        // 消费组B,不自动ack
        String redisStreamKey = "test_stream";
        redisStreamUtil.initGroup(redisStreamKey, ReadOffset.latest(), "group-b");//从尾消费
        Subscription subscription = listenerContainer.receive(Consumer.from("group-b", "consumer-bb"),
                StreamOffset.create(redisStreamKey, ReadOffset.lastConsumed()),
                new TestAsyncConsumeStreamListener("消费组B消费", "group-b", "consumer-bb"));

        listenerContainer.start();
        return subscription;
    }*/

    /**
     * 可以同时支持 独立消费 和 消费者组 消费
     * <p>
     * 可以支持动态的 增加和删除 消费者
     * <p>
     * 消费组需要预先创建出来
     *
     * @return StreamMessageListenerContainer
     */
    /*@Bean(initMethod = "start", destroyMethod = "stop")
    public StreamMessageListenerContainer<String, ObjectRecord<String, Object>> streamMessageListenerContainer() {
        AtomicInteger index = new AtomicInteger(1);
        int processors = Runtime.getRuntime().availableProcessors();
        ThreadPoolExecutor executor = new ThreadPoolExecutor(processors, processors, 0, TimeUnit.SECONDS,
                new LinkedBlockingDeque<>(), r -> {
            Thread thread = new Thread(r);
            thread.setName("async-stream-consumer-" + index.getAndIncrement());
            thread.setDaemon(true);
            return thread;
        });
        StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, ObjectRecord<String, Object>> options =
                StreamMessageListenerContainer.StreamMessageListenerContainerOptions
                        .builder()
                        // 一次最多获取多少条消息
                        .batchSize(5)
                        // 运行 Stream 的 poll task
                        .executor(executor)
                        // 可以理解为 Stream Key 的序列化方式
                        .keySerializer(RedisSerializer.string())
                        // 可以理解为 Stream 后方的字段的 key 的序列化方式
                        .hashKeySerializer(RedisSerializer.string())
                        // 可以理解为 Stream 后方的字段的 value 的序列化方式
                        .hashValueSerializer(RedisSerializer.string())
                        // Stream 中没有消息时，阻塞多长时间，需要比 `spring.redis.timeout` 的时间小 秒
                        //.pollTimeout(Duration.ofSeconds(3))
                        // ObjectRecord 时，将 对象的 filed 和 value 转换成一个 Map 比如：将Book对象转换成map
                        .objectMapper(new ObjectHashMapper())
                        // 获取消息的过程或获取到消息给具体的消息者处理的过程中，发生了异常的处理
                        .errorHandler(new CustomErrorHandler())
                        // 将发送到Stream中的Record转换成ObjectRecord，转换成具体的类型是这个地方指定的类型
                        .targetType(Object.class)
                        .build();

        StreamMessageListenerContainer<String, ObjectRecord<String, Object>> streamMessageListenerContainer =
                StreamMessageListenerContainer.create(redisConnectionFactory, options);

        String redisStreamKey = "test_stream";
        // 独立消费
        //streamMessageListenerContainer.receive(StreamOffset.fromStart(redisStreamKey),
        //new TestAsyncConsumeStreamListener2("独立消费", null, null));

        // 消费组A,消费者a（和消费者b为竞争关系） 自动ack
        // 从消费组中没有分配给消费者的消息开始消费
        redisStreamUtil.initGroup("test_stream", "group-a");
        streamMessageListenerContainer.receive(Consumer.from("group-a", "consumer-a"),
                StreamOffset.create(redisStreamKey, ReadOffset.lastConsumed()),
                new TestAsyncConsumeStreamListener2("消费组A消费","group-a", "consumer-a"));

        // 消费组A,消费者b（和消费者a为竞争关系）自动ack
        // 从消费组中没有分配给消费者的消息开始消费
        streamMessageListenerContainer.receiveAutoAck(Consumer.from("group-a", "consumer-b"),
                StreamOffset.create(redisStreamKey, ReadOffset.lastConsumed()),
                new TestAsyncConsumeStreamListener2("消费组A消费","group-a", "consumer-b"));

        // 消费组B,不自动ack
        redisStreamUtil.initGroup("test_stream", "group-b");
        streamMessageListenerContainer.receive(Consumer.from("group-b", "consumer-bb"),
                StreamOffset.create(redisStreamKey, ReadOffset.lastConsumed()),//ReadOffset.from("0-0")
                new TestAsyncConsumeStreamListener2("消费组B消费", "group-b", "consumer-bb"));
        streamMessageListenerContainer.start();
        return streamMessageListenerContainer;
    }*/
}
